package com.ls.fw.hadoop;

import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.net.URI;
import java.net.URISyntaxException;
import java.util.Enumeration;
import java.util.List;

import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.BlockLocation;
import org.apache.hadoop.fs.CommonConfigurationKeys;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FSInputStream;
import org.apache.hadoop.fs.FileContext;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.UnresolvedLinkException;
import org.apache.hadoop.hdfs.DFSClient;
import org.apache.hadoop.hdfs.DFSConfigKeys;
import org.apache.hadoop.hdfs.DFSInputStream;
import org.apache.hadoop.hdfs.DistributedFileSystem;
import org.apache.hadoop.hdfs.protocol.LocatedBlock;
import org.apache.hadoop.hdfs.protocol.LocatedBlocks;
import org.apache.hadoop.hdfs.server.namenode.NameNode;
import org.apache.hadoop.hdfs.web.WebHdfsFileSystem;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.security.AccessControlException;
import org.mortbay.jetty.InclusiveByteRange;

public class HdfsDaoImpl implements HdfsDao{
	
	final Log LOG = LogFactory.getLog(HdfsDaoImpl.class);
	
	private Configuration conf = null;
	private FileSystem fs;
	private FileContext fc;
	//private HftpFileSystem hftpfs; 
	private DFSClient dfsClient;
	private DistributedFileSystem dfs = new DistributedFileSystem();
	private WebHdfsFileSystem wfs = null;
	/**
	 * 
	 */
	public HdfsDaoImpl() {
		super();
	}

	/**
	 * @param conf
	 * @throws IOException 
	 */
	public HdfsDaoImpl(Configuration conf) throws Exception {
		super();
		this.setConf(conf);
	}

	@Override
	public Configuration getConfiguration() {
		return conf;
	}

	@Override
	public synchronized void setConf(Configuration conf) throws Exception{
		this.conf = conf;
		final String hdfsurl = conf.get(CommonConfigurationKeys.FS_DEFAULT_NAME_KEY);
		if(fs==null){
			dfs = new DistributedFileSystem();
		}
		dfs.initialize(URI.create(hdfsurl), conf);
		
		if(fs!=null){
			fs.close();
		}
	    fs = FileSystem.get(URI.create(hdfsurl), conf);
	    
	    fc = FileContext.getFileContext(conf);
	   
	    if(dfsClient!=null){
	    	dfsClient.close();
		}
	    //dfsClient = new DFSClient(URI.create(hdfsurl), conf);
	    dfsClient = new DFSClient(NameNode.getAddress(conf), conf);
	}
	
	@Override
	public synchronized WebHdfsFileSystem openWebHdfsFileSystem() throws IOException, URISyntaxException{
		if(wfs==null){
			final String wfsuri = WebHdfsFileSystem.SCHEME  + "://"
			+ conf.get(DFSConfigKeys.DFS_NAMENODE_HTTP_ADDRESS_KEY);
			wfs = (WebHdfsFileSystem)FileSystem.get(new URI(wfsuri), conf);
		}
		return wfs;
	}

//	@Override
//	public synchronized HftpFileSystem openHftpFileSystem() throws IOException, URISyntaxException{
//		 if(hftpfs==null){
//			 final String uri = "hftp://" + conf.get(DFSConfigKeys.DFS_NAMENODE_HTTP_ADDRESS_KEY);
//			 hftpfs = (HftpFileSystem)FileSystem.get(new URI(uri), conf);
//		 }
//		 return hftpfs;
//	}
//	
	
	@Override
	public void destory() {
		if(fs!=null){
			try {
				fs.close();
			} catch (IOException e) {
				LOG.info(e.getMessage(), e);
			}
			fs = null;
		}
		if(dfsClient!=null){
	    	try {
				dfsClient.close();
			} catch (IOException e) {
				LOG.info(e.getMessage(), e);
			}
			dfsClient = null;
		}
//		if(hftpfs!=null){
//	    	try {
//				hftpfs.close();
//			} catch (IOException e) {
//				LOG.info(e.getMessage(), e);
//			}
//			hftpfs = null;
//		}
		if(dfs!=null){
			try {
				dfs.close();
			} catch (IOException e) {
				LOG.info(e.getMessage(), e);
			}
		}
	}
	
	@Override
	public Path getPath(String path){
		return new Path(path);
	}
	

	@Override
	public Path[] getPaths(String... path) {
		Path[] paths = null;
		if(path!=null){
			paths = new Path[path.length];
			int index = 0;
			for (String string : path) {
				paths[index++] = this.getPath(string);
			}
		}
		return paths;
	}

	@Override
	public Path getPath(URI uri) {
		return new Path(uri);
	}

	@Override
	public Path getPath(String parent, String child) {
		return new Path(parent,child);
	}

	@Override
	public FileStatus[] list(String... path) throws IOException {
		FileStatus[] fsf = null;
		try {
			fsf = this.getFileSystem().listStatus(this.getPaths(path));
		} catch (IOException e) {
			throw e;
		}
		return fsf;
	}

	@Override
	public DFSClient getDFSClient() {
		return this.dfsClient;
	}

	@Override
	public FileContext getFileContext() {
		return this.fc;
	}

	@Override
	public FileSystem getFileSystem() {
		return this.fs;
	}
//
//	@Override
//	public HftpFileSystem getHftpFileSystem() {
//		return this.hftpfs;
//	}

	@Override
	public DistributedFileSystem getDistributedFileSystem() {
		return this.dfs;
	}
	
	public WebHdfsFileSystem getWebHdfsFileSystem(){
	    return this.wfs;
	}
	
	@Override
	public  boolean isExist(String path) throws Exception {
		FileSystem fs = null;
		boolean flag = false;
		try {
			fs = this.getFileSystem();
			Path dst = new Path(path);
			flag = fs.exists(dst);
		} catch (Exception e) {
			throw e;
		} finally {
		}
		return flag;
	}
	
	@Override
	public  FSDataInputStream getInputStream(String path)throws Exception {
		FSDataInputStream iStream = null;
		try {
			if (this.isExist(path)) {
				iStream = this.getFileSystem().open(new Path(path));
			}
		} catch (Exception e) {
			throw e;
		} finally {
			
		}
		return iStream;
	}
	
	@Override
	public  FSDataOutputStream getOutputStream(String path) throws Exception {
		FSDataOutputStream fos = null;
		FileSystem fs = null;
		try {
			fs = this.getFileSystem();
			fos = fs.create(new Path(path));
		} catch (Exception e) {
			throw e;
		} finally {
			
		}
		return fos;
	}

	
	@Override
	public  FileStatus getFileStatus(String srcPath) throws IOException {
		FileStatus fileStatus = null;
		FileSystem fs = null;
		try {
			fs = this.getFileSystem();
			final Path listpath = fs.makeQualified(new Path(srcPath));
			fileStatus = fs.getFileStatus(listpath);
		} catch (IOException e) {
			throw e;
		} finally {
		}
		return fileStatus;
	}
	
	
	@Override
	public  boolean deleteFile(String path) throws IOException {
		FileSystem fs = null;
		boolean flag = false;
		try {
			fs = this.getFileSystem();
			Path src = new Path(path);
			flag = fs.delete(src, false);
		} catch (IOException e) {
			throw e;
		} finally {
			
		}
		return flag;
	}
	
	
	@Override
	public  boolean deleteDir(String path, boolean recursive) throws IOException {
		FileSystem fs = null;
		boolean flag = false;
		try {
			fs = this.getFileSystem();
			final Path listpath = fs.makeQualified(new Path(path));
			flag = fs.delete(listpath, recursive);
		} catch (IOException e) {
			throw e;
		} finally {
			
		}
		return flag;
	}
	
	
	@Override
	public  boolean delete(String path) throws IOException {
		boolean flag = false;
		try {
			if(this.getFileStatus(path).isFile()){
				this.deleteFile(path);
			}else{
				this.deleteDir(path, true);
			}
		} catch (IOException e) {
			throw e;
		} finally {
			
		}
		return flag;
	}
	
	
	@Override
	public  boolean createNewFile(String path) throws Exception {
		boolean flag = false;
		try {
			if (isExist(path)) {
				throw new Exception("该文件已经存在！");
			} else {
				flag = this.getFileSystem().createNewFile(new Path(path));
			}
		} catch (Exception e) {
			throw e;
		} finally {
			
		}
		return flag;
	}
	
	
	@Override
	public  boolean mkdirs(String path) throws Exception {
		boolean flag = false;
		try {
			if (isExist(path)) {
				throw new Exception("该文件夹已经存在！");
			} else {
				flag = this.getFileSystem().mkdirs(new Path(path));
			}
		} catch (Exception e) {
			throw e;
		} finally {
			
		}
		return flag;
	}
	
	
	@Override
	public  boolean rename(String srcPath, String destPath)throws Exception {
		boolean flag = false;
		try {
			Path src = new Path(srcPath);
			Path dst = new Path(destPath);
			if (isExist(srcPath) && !isExist(destPath)) {
				flag = this.getFileSystem().rename(src, dst);
			}
		} catch (IOException e) {
			throw e;
		} catch (Exception e1) {
			throw e1;
		} finally {
			
		}
		return flag;
	}
	
	/**
	 * 向srcPath添加destPath中的内容
	 * @param srcPath
	 * @param destPath
	 * @throws Exception
	 */
	@Override
	public  void append(String srcPath, String destPath) throws Exception {
		FSDataOutputStream fos = null;
		InputStream is = null;
		try {
			FileSystem hdfs = this.getFileSystem();
			Path dst = new Path(destPath);
			is = new FileInputStream(srcPath);
			fos = hdfs.append(dst);
			IOUtils.copyBytes(is, fos, 1024, true);
			fos.flush();
		} catch (Exception e) {
			throw e;
		} finally {
			if (is != null) {
				try {
					is.close();
				} catch (Exception e) {
					e.printStackTrace();
				}
			}
			IOUtils.closeStream(fos);
		}
	}

	@Override
	public void down(String srcPath, String destPath)throws Exception {
		try {
			if (isExist(srcPath)) {
				Path src = new Path(srcPath);
				Path dst = new Path(destPath);
				this.getFileSystem().copyToLocalFile(src, dst);
			}else{
				throw new Exception(srcPath+"不存在！");
			}
		} catch (Exception e) {
			throw e;
		} finally {
			
		}
	}
	
	@Override
	public  void upload(String srcPath, String destDirOrdestPath)throws Exception {
		try {
			Path src = new Path(srcPath);
			Path dst = new Path(destDirOrdestPath);
			this.getFileSystem().copyFromLocalFile(src, dst);
		} catch (Exception e) {
			throw e;
		} finally {
		}
	}
	
	
	@Override
	public  void upload(InputStream inStream, String destPath)throws Exception {
		FSDataOutputStream fos = null;
		try {
			fos = this.getOutputStream(destPath);
			IOUtils.copyBytes(inStream, fos, 1024, true);
			fos.flush();
		} catch (Exception e) {
			throw e;
		} finally {
			IOUtils.closeStream(fos);
		}
	}
	
	@Override
	@SuppressWarnings("unchecked")
	public   void down(HttpServletRequest request,
			HttpServletResponse response, String path, String name)
			throws IOException {
		DFSInputStream in = null;
		OutputStream out = null;
		Enumeration<String> reqRanges = request.getHeaders("Range");
		if (reqRanges != null && !reqRanges.hasMoreElements()) {
			reqRanges = null;
		}
		try {
			in = this.getDFSClient().open(path);
			out = response.getOutputStream();
			final long fileLen = in.getFileLength();
			if (reqRanges != null) {
				List<InclusiveByteRange> ranges = InclusiveByteRange
						.satisfiableRanges(reqRanges, fileLen);
				this.sendPartialData(in, out, response, fileLen, ranges);
			} else {
				// No ranges, so send entire file
				response.setHeader("Content-Disposition",
						"attachment; filename=\"" + name + "\"");
				response.setContentType("application/octet-stream");
				response.setHeader("Content-Length", "" + fileLen);
				this.copyFromOffset(in, out, 0L, fileLen);
			}
		} catch (IOException ioe) {
			throw ioe;
		} finally {
			if (in != null) {
				try {
					in.close();
					in = null;
				} catch (Exception e) {
				}
			}
			if (out != null) {
				try {
					out.close();
					out = null;
				} catch (Exception e) {
				}
			}
		}
	}

	/**
	 * Send a partial content response with the given range. If there are no
	 * satisfiable ranges, or if multiple ranges are requested, which is
	 * unsupported, respond with range not satisfiable.
	 * 
	 * @param in
	 *            stream to read from
	 * @param out
	 *            stream to write to
	 * @param response
	 *            http response to use
	 * @param contentLength
	 *            for the response header
	 * @param ranges
	 *            to write to respond with
	 * @throws IOException
	 *             on error sending the response
	 */
	private   void sendPartialData(FSInputStream in, OutputStream out,
			HttpServletResponse response, long contentLength,
			List<InclusiveByteRange> ranges) throws IOException {
		if (ranges == null || ranges.size() != 1) {
			response.setContentLength(0);
			response
					.setStatus(HttpServletResponse.SC_REQUESTED_RANGE_NOT_SATISFIABLE);
			response.setHeader("Content-Range", InclusiveByteRange
					.to416HeaderRangeString(contentLength));
		} else {
			InclusiveByteRange singleSatisfiableRange = ranges.get(0);
			long singleLength = singleSatisfiableRange.getSize(contentLength);
			response.setStatus(HttpServletResponse.SC_PARTIAL_CONTENT);
			response.setHeader("Content-Range", singleSatisfiableRange
					.toHeaderRangeString(contentLength));
			copyFromOffset(in, out, singleSatisfiableRange
					.getFirst(contentLength), singleLength);
		}
	}

	/* Copy count bytes at the given offset from one stream to another */
	private   void copyFromOffset(FSInputStream in, OutputStream out,
			long offset, long count) throws IOException {
		in.seek(offset);
		IOUtils.copyBytes(in, out, count, false);
	}
	
	
	@Override
	public BlockLocation[] getBlockLocation(FileStatus fileStatus) throws IOException{
		BlockLocation[] blockLocations = this.getDistributedFileSystem().getFileBlockLocations(fileStatus, 0, fileStatus.getLen());
		return blockLocations;
	}
	
	@Override
	public BlockLocation[] getBlockLocation(String path) throws IOException{
		return this.getBlockLocation(this.getFileStatus(path));
	}
	
	@Override
	public List<LocatedBlock> getLocatedBlock(String path) throws AccessControlException, FileNotFoundException, UnresolvedLinkException, IOException{
		LocatedBlocks locations = this.getDFSClient().getNamenode().getBlockLocations(
				path, 0, Long.MAX_VALUE);
		List<LocatedBlock> blocks = locations.getLocatedBlocks();
		return blocks;
	}
}
